package cn.jly.bigdata.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @author lanyangji
 * @date 2019/11/25 20:09
 */
object Practice {

  def main(args: Array[String]): Unit = {

    // 数据样本
    // 时间戳  省 市 用户  广告（一条记录为一次广告点击）
    // 1516609143867 6 7 64 16
    // 1516609143869 9 4 75 18
    // 1516609143869 1 7 87 12


    // 统计出每一个省份广告被点击次数的top3

    // 1. 初始化spark配置信息并建立spark的连接
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("practice")
    val sparkContext: SparkContext = new SparkContext(sparkConf)

    // 2. 读取数据生成RDD
    val lines: RDD[String] = sparkContext.textFile("d:/agent.log")

    // 3. 按照最小的粒度聚合，这边只考虑省份和广告，以及点击了1次: ((省份，广告),1)
    val provinceAdToOne: RDD[((String, String), Int)] = lines.map {
      x =>
        val fields: Array[String] = x.split(" ")
        ((fields(1), fields(4)), 1)
    }

    //println(provinceAdToOne.collect.mkString(","))

    // 4. 计算每个省份每个广告被点击的次数 （（省份，广告）, sum）
    val provinceAdToSum: RDD[((String, String), Int)] = provinceAdToOne.reduceByKey(_ + _)

    //println(provinceAdToSum.collect.mkString(","))

    // 5. 将省份作为key，广告和点击次数作为value: (省份，（广告，sum）)
    val provinceToAdSum: RDD[(String, (String, Int))] = provinceAdToSum.map(x => (x._1._1, (x._1._2, x._2)))

    //println(provinceToAdSum.collect.mkString(","))

    // 6. 将同一个省份的所有广告进行聚合：（省份，List((广告，sum),(广告2，sum2)...)）
    val provinceGroup: RDD[(String, Iterable[(String, Int)])] = provinceToAdSum.groupByKey

    //println(provinceGroup.collect.mkString(","))

    // 7. 对同一个省份的所有广告的集合进行排序并取前三条，排序规则为广告的点击数
    val provinceAdTop3: RDD[(String, List[(String, Int)])] = provinceGroup.mapValues {
      x => x.toList.sortWith((elem1, elem2) => elem1._2 > elem2._2).take(3)
    }

    provinceAdTop3.collect.foreach(println)

    // 关闭与spark的连接
    sparkContext.stop()
  }
}
